package com.zhiyuancode.mqack.worker.recorder;


import com.zhiyuancode.mqack.message.base.MqResendErrorCode;
import com.zhiyuancode.mqack.message.base.MqResendRuntimeException;
import com.zhiyuancode.mqack.message.base.addressAndMessage.AddressAndMessage4String;
import com.zhiyuancode.mqack.message.base.message.BasicMessage;
import com.zhiyuancode.mqack.worker.RedisOperationsContext;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.RedisConnectionFailureException;

import java.io.IOException;
import java.util.*;

import static com.zhiyuancode.mqack.worker.Constants.*;


/**
 * 消息体有多个地址
 *
 * @author 杨志远
 * email faryangsh@163.com
 * Date    2018-05-02
 */
@Slf4j
public class Recorder4Json extends BaseRecorder {
    Class messageClass;
    protected final int resendMaxCount;
    protected final long resendTimeOutSecond;

    /**
     * @param redisOperationsContext redis操作操作对象
     * @param messageClass           消息类
     * @param resendTimeOutSecond    容忍消息ACK间隔时间(秒)
     */
    public Recorder4Json(RedisOperationsContext redisOperationsContext, Class messageClass, long resendTimeOutSecond, int resendMaxCount) {
        this.hashOperations = redisOperationsContext.getHashOperations();
        this.zSetOperations = redisOperationsContext.getZSetOperations();
        this.messageClass = messageClass;
        this.resendTimeOutSecond = resendTimeOutSecond;
        this.resendMaxCount = resendMaxCount;
    }

    public Recorder4Json(RedisOperationsContext redisOperationsContext, long resendTimeOutSecond, int resendMaxCount) {
        this.hashOperations = redisOperationsContext.getHashOperations();
        this.zSetOperations = redisOperationsContext.getZSetOperations();
        this.resendTimeOutSecond = resendTimeOutSecond;
        this.resendMaxCount = resendMaxCount;


    }

    public Class getMessageClass() {
        return messageClass;
    }

    /**
     * 保存已发的MQ记录
     *
     * @param
     */
    public void saveMessageRecord(BasicMessage message) {
        try {
            // 记录该消息的msgId
            zSetOperations.add(RESEND_MSG_ID_COL, message.acquireMsgId(), System.currentTimeMillis());

            // 记录该消息的内容、地址、class类型
            Map<String, String> mqMsgMap = new HashMap<>(4);
            mqMsgMap.put(RESEND_MSG_ADDRESS, message.acquireMsgAddress());
            mqMsgMap.put(RESEND_MSG_CONTENT, message.acquireMsgContent());
            mqMsgMap.put(RESEND_MSG_TYPE, message.acquireMqType().name());
            mqMsgMap.put(RESEND_MSG_CONTENT_CLASS, message.getClass().getName());
            mqMsgMap.put(RESEND_COUNT, "0");
            hashOperations.putAll(RESEND_MSG_PREFIX + message.acquireMsgId(), mqMsgMap);
        } catch (RedisConnectionFailureException e) {
            log.error("failed save message record: msgId={}, e={}", message.acquireMsgId(), e.getMessage());
            throw new MqResendRuntimeException(MqResendErrorCode.REDIS_CONNECTION_EXCEPTION, e);
        } catch (Throwable e) {
            log.error("failed save message record: msgId={}, e={}", message.acquireMsgId(), e.getMessage());
            throw new MqResendRuntimeException(MqResendErrorCode.UNKNOW_EXCEPTION, e);
        }
    }

    /**
     * 保存已发的MQ记录
     * - 用于消息体一个，但消息地址多个的消息
     *
     * @param
     */
    public void saveMessageRecord4MutilMsgAddress(BasicMessage message, String msgAddress) {
        try {
            // 记录该消息的msgId
            zSetOperations.add(RESEND_MSG_ID_COL, message.acquireMsgId(), System.currentTimeMillis());

            // 记录该消息的内容、地址、class类型
            Map<String, String> mqMsgMap = new HashMap<>(4);
            mqMsgMap.put(RESEND_MSG_ADDRESS, msgAddress);
            mqMsgMap.put(RESEND_MSG_CONTENT, message.acquireMsgContent());
            mqMsgMap.put(RESEND_MSG_TYPE, message.acquireMqType().name());
            mqMsgMap.put(RESEND_MSG_CONTENT_CLASS, message.getClass().getName());
            hashOperations.putAll(RESEND_MSG_PREFIX + message.acquireMsgId(), mqMsgMap);
        } catch (RedisConnectionFailureException e) {
            log.error("failed save message record: msgId={}, e={}", message.acquireMsgId(), e.getMessage());
            throw new MqResendRuntimeException(MqResendErrorCode.REDIS_CONNECTION_EXCEPTION, e);
        } catch (Throwable e) {
            log.error("failed save message record: msgId={}, e={}", message.acquireMsgId(), e.getMessage());
            throw new MqResendRuntimeException(MqResendErrorCode.UNKNOW_EXCEPTION, e);
        }
    }

    /**
     * 查询待重发的消息
     *
     * @return
     */
    public List<AddressAndMessage4String> queryResendMessage() throws IOException {
        Set<String> resendMsgId = getResendMsgIds();
        if (resendMsgId == null || resendMsgId.size() == 0) {
            return null;
        } else {
            List<AddressAndMessage4String> mqList = getAddressAndMessage4Resend(resendMsgId);
            return mqList;
        }
    }

    /**
     * 查询超过重发限制次数的消息
     *
     * @return
     */
    public List<String> queryOverMaxCountMsgIds() throws IOException {
        Set<String> resendMsgId = getResendMsgIds();
        return resendMsgId == null || resendMsgId.size() == 0 ? null : getOverMaxCountMsgIds(resendMsgId);
    }

    /**
     * 清除超过重发限制次数的消息记录
     *
     * @return
     */
    public List<String> clearOverMaxCountMsgIdRecords() throws IOException {
        List<String> overMaxCountMsgIds = queryOverMaxCountMsgIds();
        for (String msgId : overMaxCountMsgIds) {
            deleteMessageRecord(msgId);
        }
        return overMaxCountMsgIds;
    }


    private List<String> getOverMaxCountMsgIds(Set<String> resendMsgId) {
        List<String> msgIds = new ArrayList<>();
        for (String msgId : resendMsgId) {
            Map<String, String> msgContent = hashOperations.entries(RESEND_MSG_PREFIX + msgId);

            // 是否超过最大重发次数
            int resend_count = Integer.parseInt(msgContent.get(RESEND_COUNT));
            if (resend_count < resendMaxCount) {
                continue;
            }
            // 不是该class类型的消息，不计数
            if (!msgContent.get(RESEND_MSG_CONTENT_CLASS).equals(messageClass.getName())) {
                continue;
            }

            msgIds.add(msgId);
        }
        return msgIds;
    }

    private List<AddressAndMessage4String> getAddressAndMessage4Resend(Set<String> resendMsgId) {
        List<AddressAndMessage4String> mqList = new ArrayList<>();
        // 对所有超时的消息记录进行重发
        for (String msgId : resendMsgId) {
            Map<String, String> msgContent = hashOperations.entries(RESEND_MSG_PREFIX + msgId);
            String mqAddress = msgContent.get(RESEND_MSG_ADDRESS);
            String mqContent = msgContent.get(RESEND_MSG_CONTENT);
            String mqType = msgContent.get(RESEND_MSG_TYPE);

            // 校验是否达到最大重发次数
            int resend_count = Integer.parseInt(msgContent.get(RESEND_COUNT));
            if (resend_count >= resendMaxCount) {
                log.info("the resending count of mq have reached max num, msgId={} , count={}", msgId, resend_count);
                continue;
            }
            // 不是该class类型的消息，不做重发
            if (!msgContent.get(RESEND_MSG_CONTENT_CLASS).equals(messageClass.getName())) {
                continue;
            }

            resend_count++;
            msgContent.put(RESEND_COUNT, String.valueOf(resend_count));

            AddressAndMessage4String mqAddressAndMessage;
            mqAddressAndMessage = AddressAndMessage4String.builder()
                    .mqAddress(mqAddress)
                    .message(mqContent)
                    .mqType(mqType)
                    .build();

            mqList.add(mqAddressAndMessage);

            // 保存重发记录(重发次数+1)
            hashOperations.putAll(RESEND_MSG_PREFIX + msgId, msgContent);
        }
        return mqList;
    }


    /**
     * 获取超时重发的消息Id
     *
     * @return 消息Id集合
     */
    private Set<String> getResendMsgIds() {
        long expireTime = System.currentTimeMillis() - resendTimeOutSecond * 1000;
        // 获取超过容忍时间外的消息记录
        return zSetOperations.rangeByScore(RESEND_MSG_ID_COL, 0, expireTime);
    }
}
